package com.lzj.pagerank;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/**
 * @author Administrator
 * //页面对应关系及老的pr值
 * //a	1.0 b d
 * //新的投票值
 * //a	0.5
 * //最终写出的结果
 * //a 0.5 c
 * //计算收敛差值 0.5-1.0= -0.5
 */
public class PageRankReducer extends Reducer<Text, Text, Text, Text> {
    protected void reduce(Text key, Iterable<Text> iterable, Context context) throws IOException, InterruptedException {

        //定义变量存放新的PR值(阻尼之前)
        double sum = 0.0;
        //获取原来的节点
        PageNode sourceNode = null;
        for (Text i : iterable) {
            //创建新的节点
            PageNode node = PageNode.fromMR(i.toString());
            //判断是老的映射关系还是新的PR值
            if (node.containsAdjacentNodes()) {
                sourceNode = node;
            } else {
                sum = sum + node.getPageRank();
            }
        }

        // 基于阻尼系数为重新计算PR值
        double newPR = (0.15 / 4.0) + (0.85 * sum);
        System.out.println(key + "*********** new pageRank value is " + newPR);

        //把新的pr值和计算之前的pr比较（获取收敛值）
        double d = newPR - sourceNode.getPageRank();
        //保留四位有效数字,然后取绝对值
        int j = Math.abs((int) (d * 1000.0));
        context.getCounter(Mycounter.CONVERGENCESUM).increment(j);

        //将当前网站的PR值写出
        sourceNode.setPageRank(newPR);
        context.write(key, new Text(sourceNode.toString()));
    }
}